package prjCode;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;


public class CalcSim {

	static private Text one = new Text("1");
	static private Text two = new Text("2");

	public static class MapClass extends Mapper<LongWritable, Text, TextTaggedKey, Text> {

		private Text primKey                    = new Text();
		private Text Line                       = new Text();
		private TextTaggedKey dKey              = new TextTaggedKey();
		private int i,j;
		private String trip;
		private String index;
		private String str;

		@Override
		public void map(LongWritable key, Text value, Context context) throws IOException,  InterruptedException {

			str = value.toString();
			index      = str.substring(0, str.indexOf("\t")); // % between decomposed by   /%/attack/10/	10 -> [ % between decomposed by   , %, attack, 10, 	10]
			trip       = str.substring(str.indexOf("\t") + 1);
			Line.set(trip);
			i = Integer.parseInt(index);

			primKey.set(String.valueOf(i));
			dKey.setKey(primKey);

			dKey.setTag(one);
			context.write(dKey, Line);

			for(j=i-1; j>0; j--)
			{
				primKey.set(String.valueOf(j));
				dKey.setKey(primKey);
				dKey.setTag(two);
				context.write(dKey, Line);
			}

		}

	}

	public static class ReduceClass extends Reducer<TextTaggedKey ,Text, Text ,Text> {



		private Text retVal          = new Text();
		private Text retRes          = new Text();
		IntWritable val              = new IntWritable();
		private double sumX;
		private double sumY;
		private double simX;
		private double simY;
		private double res;
		private String mainLine        = "";
		private String line            = "";
		private String[] aLine;
		private String[] aMainLine;
		private String[] slotXM;
		private String[] slotYM;
		private String[] slotXL;
		private String[] slotYL;
		private int numOfCommFeature, i, j, comp, lx, ly;

		@Override
		public void reduce(TextTaggedKey key, Iterable<Text> values, Context context) throws IOException,  InterruptedException {

			final Configuration conf = context.getConfiguration();
			
			float f = conf.getFloat("min",0);

			if(key.getTag().equals(one))
			{
				mainLine = values.iterator().next().toString();
				aMainLine = mainLine.split("/");
				slotXM    = aMainLine[1].split(" ");
				slotYM    = aMainLine[2].split(" ");
				lx = slotXM.length/2;
				ly = slotYM.length/2;
			}
			else{


				while (values.iterator().hasNext()) {

					sumX = sumY = i = j = 0;

					numOfCommFeature = 0;

					line = values.iterator().next().toString();
					aLine = line.split("/");
					slotXL    = aLine[1].split(" ");
					slotYL    = aLine[2].split(" ");

					while(i < slotXM.length && j < slotXL.length)
					{
						comp = slotXM[i].compareTo(slotXL[j]);

						if (comp == 0)
						{
							numOfCommFeature++;
							sumX += (Double.parseDouble(slotXM[i + 1]) + Double.parseDouble(slotXL[j + 1]));
							i += 2;
							j += 2;
						}
						else if (comp < 0)
						{
							i += 2;
						}
						else
						{
							j += 2;
						}
					}

					
					if (f <= (((double)numOfCommFeature)/(lx + slotXL.length/2)))
					{
						numOfCommFeature = i = j = 0;

						while(i < slotYM.length && j < slotYL.length)
						{
							comp = slotYM[i].compareTo(slotYL[j]);

							if (comp == 0)
							{
								numOfCommFeature++;
								sumY += (Double.parseDouble(slotYM[i + 1]) + Double.parseDouble(slotYL[j + 1]));
								i += 2;
								j += 2;
							}
							else if (comp < 0)
							{
								i += 2;
							}
							else
							{
								j += 2;
							}
						}

						if (f <= (((double)numOfCommFeature)/(ly + slotYL.length/2)))
						{
							simX = sumX/(Double.valueOf(aLine[3]) + Double.valueOf(aMainLine[3])) ;
							simY = sumY/(Double.valueOf(aLine[4]) + Double.valueOf(aMainLine[4])) ;
							if (simX >= 0 && simY >= 0)
								res  = Math.sqrt(simX * simY);
							else
								res  = simX * simY;
							retVal.set(aMainLine[0] + "/" + aLine[0]);
							retRes.set(String.valueOf(res));
							context.write(retRes, retVal);
						}

					}


				}
			}

		}

	}


	public static class PartitionerClass extends Partitioner<TextTaggedKey,Text> {
		// ensure that keys with same key are directed to the same reducer
		@Override
		public int getPartition(TextTaggedKey key,Text value, int numPartitions) {
			return  Math.abs(key.getKey().hashCode()) % numPartitions;
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		conf.set("min"   , (args[2]));
		Job job = new Job(conf, "Sim");
		job.setJarByClass(CalcSim.class);
		job.setMapperClass(CalcSim.MapClass.class);
		job.setPartitionerClass(CalcSim.PartitionerClass.class);
		job.setReducerClass(CalcSim.ReduceClass.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setMapOutputKeyClass(TextTaggedKey.class);
		job.setMapOutputValueClass(Text.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}
